package day06;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;
import po.OrderEvent;

/**
 * @Description: 实时对账 (订单根据id匹配消费)
 * @Author ZYX
 * @Date 2022/2/15 19:55
 * @Version 1.0
 */
public class Demo02 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<OrderEvent> orderEventStream = env.fromElements(
                OrderEvent.of("order-1", "pay", 1000L),
                OrderEvent.of("order-1", "pay", 2000L)
        ).assignTimestampsAndWatermarks(
                WatermarkStrategy.<OrderEvent>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<OrderEvent>() {
                            @Override
                            public long extractTimestamp(OrderEvent orderEvent, long l) {
                                return orderEvent.timestamp;
                            }
                        })
        );

        SingleOutputStreamOperator<OrderEvent> wechatStream = env.fromElements(
                OrderEvent.of("order-1", "wechat", 30000L),
                OrderEvent.of("order-3", "wechat", 4000L)
        ).assignTimestampsAndWatermarks(
                WatermarkStrategy.<OrderEvent>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<OrderEvent>() {
                            @Override
                            public long extractTimestamp(OrderEvent orderEvent, long l) {
                                return orderEvent.timestamp;
                            }
                        })
        );

        orderEventStream.keyBy(obj -> obj.orderId)
                .connect(wechatStream.keyBy(obj -> obj.orderId))
                .process(new MatchFunction())
                .print();


        env.execute();
    }

    public static class MatchFunction extends CoProcessFunction<OrderEvent,OrderEvent,String>{
        private ValueState<OrderEvent> orderState;
        private ValueState<OrderEvent> wechatState;

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            orderState = getRuntimeContext().getState(new ValueStateDescriptor<OrderEvent>("order", Types.POJO(OrderEvent.class)));
            wechatState = getRuntimeContext().getState(new ValueStateDescriptor<OrderEvent>("wechat", Types.POJO(OrderEvent.class)));

        }

        @Override
        public void processElement1(OrderEvent orderEvent, Context context, Collector<String> out) throws Exception {
            if (wechatState.value() == null){
                // 下订单order事件先到达，因为如果weixin事件先到达，那么weixinState就不为空了
                orderState.update(orderEvent);
                context.timerService().registerEventTimeTimer(orderEvent.timestamp + 5000L);
            }else {
                out.collect("订单id是："+orderEvent.orderId+"对账成功，wechat事件先到达");
                wechatState.clear();
            }
        }

        @Override
        public void processElement2(OrderEvent wechatEvent, Context context, Collector<String> out) throws Exception {
            if (orderState.value() == null){
                wechatState.update(wechatEvent);
                context.timerService().registerEventTimeTimer(wechatEvent.timestamp+5000L);
            }else {
                out.collect("订单id是"+wechatEvent.orderId+"对账成功，order事件先到达");
                orderState.clear();
            }
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            super.onTimer(timestamp, ctx, out);
            if (orderState.value() != null) {
                out.collect("订单ID" + orderState.value().orderId + "对账失败，weixin事件5s内未到达");
                orderState.clear();
            }
            if (wechatState.value() != null) {
                out.collect("订单ID" + wechatState.value().orderId + "对账失败，order事件5s内未到达");
                wechatState.clear();
            }
        }
    }

}
